package redis

import (
	"context"
	"fmt"
	"github.com/go-mysql-org/go-mysql/canal"
	"github.com/go-mysql-org/go-mysql/mysql"
	"github.com/go-mysql-org/go-mysql/schema"
	"interview/pkg/global"
	"interview/pkg/utils"
)

const (
	idName        = "id"
	deletedAtName = "deleted_at"
	casbinName    = "go_step_tb_casbin_rule"
)

// RowChange MySQL数据行发生变化时，同步数据到Redis中
func RowChange(e *canal.RowsEvent) {
	// 数据库名和表名
	database := e.Table.Schema
	table := e.Table.Name

	// 默认用id作为主键查找id的索引位置
	idIdx := -1
	deletedAtIdx := -1
	for i, c := range e.Table.Columns {
		if c.Name == idName {
			idIdx = i
		}
		if c.Name == deletedAtName {
			deletedAtIdx = i
		}
		if idIdx >= 0 && deletedAtIdx >= 0 {
			break
		}
	}
	// key: {database}_{table}
	cacheKey := fmt.Sprintf("%s_%s", database, table)
	// 读取Redis历史数据
	oldRows, err := global.Redis.Get(context.Background(), cacheKey).Result()
	newRows := make([]map[string]interface{}, 0)
	changeRows := make([][]interface{}, 0)
	if err == nil {
		utils.JsonString2Struct(oldRows, &newRows)
	}
	for i, v := range e.Rows {
		for j, v2 := range v {
			if u, ok := v2.([]uint8); ok {
				e.Rows[i][j] = string(u)
			}
		}
	}
	utils.Struct2StructByJson(e.Rows, &changeRows)

	// 根据事件类型进行相应处理
	switch e.Action {
	case canal.InsertAction:
		newRows = append(newRows, getRow(changeRows[0], e.Table))
		break
	case canal.UpdateAction:
		for i, row := range newRows {
			if idIdx >= 0 && row[idName] == changeRows[0][idIdx] {
				// gorm软删除之后清理redis对应缓存
				if deletedAtIdx >= 0 && changeRows[0][deletedAtIdx] == nil && changeRows[1][deletedAtIdx] != nil {
					newRows = append(newRows[:i], newRows[i+1:]...)
					break
				}
				// 替换原有元素
				// changeRows[0]表示改变前的元素, changeRows[1]表示改变后的元素
				newRows[i] = getRow(changeRows[1], e.Table)
				break
			}
		}
		break
	case canal.DeleteAction:
		for i, row := range newRows {
			if row[idName] == changeRows[0][idIdx] {
				newRows = append(newRows[:i], newRows[i+1:]...)
				break
			}
		}
		break
	}

	if cacheKey == casbinName {
		casbinList := make([]string, 0)
		for _, row := range newRows {
			casbinList = append(casbinList, utils.Struct2JsonString(row))
		}
		if err = global.Redis.LPush(context.Background(), cacheKey, casbinList).Err(); err != nil {
			global.Logger.Error("同步 [casbin] binlog增量数据到redis失败：", err, e)
		}
	} else {
		if err = global.Redis.Set(context.Background(), cacheKey, utils.Struct2JsonString(newRows), 0).Err(); err != nil {
			global.Logger.Error("同步binlog增量数据到redis失败：", err, e)
		}
	}
}

// GetCurrentPos 获取当前日志位置
func GetCurrentPos() (pos mysql.Position) {
	result, err := global.Redis.Get(context.Background(), global.Conf.Redis.BinlogPos).Result()
	if err == nil {
		utils.JsonString2Struct(result, &pos)
	}
	return
}

// PosChange MySQL日志位置发生变化
func PosChange(position mysql.Position) {
	err := global.Redis.Set(context.Background(), global.Conf.Redis.BinlogPos, utils.Struct2JsonString(position), 0).Err()
	if err != nil {
		global.Logger.Error("同步binlog当前位置到redis失败：", err, position)
	}
}

// 获取一行的字段与数据值
func getRow(data []interface{}, table *schema.Table) map[string]interface{} {
	row := make(map[string]interface{}, 0)
	count := len(data)
	for i, c := range table.Columns {
		var item interface{}
		if i < count {
			// canal没有对tinyint(1)做bool转换, 需要手动转换
			if c.RawType == "tinyint(1)" {
				item = false
				switch data[i].(type) {
				// canal中的tinyint(1)为float64格式
				case float64:
					if int(data[i].(float64)) == 1 {
						item = true
					}
					break
				}
			} else {
				item = data[i]
			}
			row[utils.ToLowerCamelCase(c.Name)] = item
		}
	}
	if count != len(table.Columns) {
		global.Logger.Warnf("数据字段可能不匹配, columns: %v, data: %v\n", table.Columns, data)
	}
	return row
}
